package xyz.zhouxf.rpc.transport.netty.client;

import io.netty.channel.*;

import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.Promise;
import lombok.extern.slf4j.Slf4j;
import xyz.zhouxf.rpc.entity.RpcRequest;
import xyz.zhouxf.rpc.entity.RpcResponse;
import xyz.zhouxf.rpc.factory.SingletonFactory;
import xyz.zhouxf.rpc.loadbalance.LoadBalancer;
import xyz.zhouxf.rpc.loadbalance.impl.RandomLoadBalancer;
import xyz.zhouxf.rpc.registry.ServiceDiscovery;
import xyz.zhouxf.rpc.registry.nacos.NacosServiceDiscovery;
import xyz.zhouxf.rpc.serializer.RpcSerializer;
import xyz.zhouxf.rpc.transport.RpcClient;

import java.net.InetSocketAddress;

/**
 * @author zhouxf
 * @date 2022/8/20
 */
@Slf4j
public class NettyClient implements RpcClient {

    private final ServiceDiscovery serviceDiscovery;

    private final RpcSerializer serializer;

    private final WaitRequests waitRequests;

    public NettyClient() {
        this(RpcSerializer.getRpcSerializer(RpcSerializer.DEFAULT_SERIALIZER),
                new RandomLoadBalancer());
    }
    public NettyClient(RpcSerializer serializer) {
        this(serializer, new RandomLoadBalancer());
    }
    public NettyClient(LoadBalancer loadBalancer) {
        this(RpcSerializer.getRpcSerializer(RpcSerializer.DEFAULT_SERIALIZER),
                loadBalancer);
    }
    public NettyClient(RpcSerializer serializer, LoadBalancer loadBalancer) {
        this.serviceDiscovery = new NacosServiceDiscovery(loadBalancer);
        this.serializer = serializer;
        this.waitRequests = SingletonFactory.getInstance(WaitRequests.class);
    }

    @Override
    public Object sendRequest(RpcRequest request) {
        InetSocketAddress inetSocketAddress = serviceDiscovery.findService(request.getInterfaceName());
        Channel channel = ChannelManager.getChannel(inetSocketAddress, serializer);
        channel.writeAndFlush(request);
        Promise<RpcResponse> promise = new DefaultPromise<>(channel.eventLoop());
        waitRequests.put(request.getRequestId(), promise);
        try {
            promise.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        if(promise.isSuccess()) {
            return promise.getNow();
        }else {
            throw new RuntimeException(promise.cause());
        }
    }
}
